package com.yequan.rocketmq.service.impl;

import com.alibaba.fastjson.JSON;
import com.yequan.rocketmq.service.RocketMQService;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * @author yq
 * @date 2020/09/10
 */
public class RocketMQAsyncServiceImpl implements RocketMQService {

    private RocketMQProducer producer;

    @Override
    public void sendMessage(String topic, String tag, byte[] body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(topic, tag, body);
        producer.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
    }

    @Override
    public void sendMessage(String topic, String tag, String msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
    }

    @Override
    public void sendMessage(String topic, String tag, Object msgObj) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tag, JSON.toJSONString(msgObj).getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
    }

    public void setProducer(RocketMQProducer producer) {
        this.producer = producer;
    }

    public RocketMQProducer getProducer() {
        return producer;
    }
}
